package com.weishao.learn;

import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.logging.KettleLogStore;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransHopMeta;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepIOMetaInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.steps.mergerows.MergeRowsMeta;
import org.pentaho.di.trans.steps.synchronizeaftermerge.SynchronizeAfterMergeMeta;
import org.pentaho.di.trans.steps.tableinput.TableInputMeta;

/**
 * 基于Kettle的增量(变化量)"数据同步"组件SynchronizeAfterMergeMeta
 * 
 * 参考地址：
 *  https://www.it610.com/article/2050194.htm
 *  https://blog.csdn.net/iteye_17500/article/details/82617906
 *  https://blog.csdn.net/maixiaochai/article/details/83795703
 *  https://blog.csdn.net/JpHold/article/details/89309023
 *  
 *  备注：测试使用数据样例：
 *  ==================================
 *  来源表结构
 *  
DROP TABLE IF EXISTS `DEPT_1`;
CREATE TABLE `DEPT_1` (
  `DEPTNO` varchar(128) NOT NULL,
  `NAME` varchar(128) DEFAULT NULL,
  PRIMARY KEY (`DEPTNO`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

CREATE TABLE `DEPT_2` (
  `DEPTNO` varchar(128) NOT NULL,
  `NAME` varchar(128) DEFAULT NULL,
  PRIMARY KEY (`DEPTNO`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

INSERT INTO `test_for_tang`.`DEPT_2` (`DEPTNO`, `NAME`) VALUES ('1', 'a');
INSERT INTO `test_for_tang`.`DEPT_2` (`DEPTNO`, `NAME`) VALUES ('2', 'b');
INSERT INTO `test_for_tang`.`DEPT_2` (`DEPTNO`, `NAME`) VALUES ('3', 'c');
INSERT INTO `test_for_tang`.`DEPT_2` (`DEPTNO`, `NAME`) VALUES ('4', 'd');
INSERT INTO `test_for_tang`.`DEPT_2` (`DEPTNO`, `NAME`) VALUES ('5', 'e');

CREATE TABLE `DEPT_OTHER` (
  `DEPTNO` varchar(128) NOT NULL,
  `NAME` varchar(128) DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

 * 执行测试第一次
 *
INSERT INTO `test_for_tang`.`DEPT_2` (`DEPTNO`, `NAME`) VALUES ('6', 'x');

 * 执行测试第一次
 * @author tang
 *
 */
public class SyncAfterMergeApplication {

	public static void main(String[] args) throws KettleException {
		String oldInputSql="SELECT DEPTNO,`NAME` FROM `test_for_tang`.`DEPT_1`";
		String newInputSql="SELECT DEPTNO,`NAME` FROM `test_for_tang`.`DEPT_2`";
		//比较的字段key，通常为表的主键字段
		String fieldsPrimaryKey[]= {"DEPTNO"};
		//比较的字段value，通常为表中除主键外的其他字段
		String fieldsCompareValue[]= {"NAME"};
		String targetSchemaName="test_for_tang";
		String targetTableName="DEPT_OTHER";
		String[] updateLookup = {"DEPTNO","NAME"} ;
 		String [] updateStream = {"DEPTNO","NAME"};
 		Boolean [] updateOrNot= {false,true};//主键字段不更新false,其余字段更新true
		
		KettleEnvironment.init();
		
		TransMeta transMeta = new TransMeta();
		transMeta.setName("Transformation");// 设置转化的名称
		
		DatabaseMeta sourcedb = new DatabaseMeta("source", "mysql", "Native(JDBC)", "172.17.207.210", "test_for_tang",	"3306", "tangyibo", "tangyibo");
		DatabaseMeta targetdb = new DatabaseMeta("source", "mysql", "Native(JDBC)", "172.17.207.210", "test_for_tang",	"3306", "tangyibo", "tangyibo");
		sourcedb.setQuoteAllFields(true);
		sourcedb.addExtraOption(sourcedb.getPluginId(), "characterEncoding", "utf8"); 
		targetdb.setQuoteAllFields(true);
		targetdb.addExtraOption(targetdb.getPluginId(), "characterEncoding", "utf8"); 
		
		transMeta.addDatabase(sourcedb);
		transMeta.addDatabase(targetdb);
		
		//表1输入（OLD）
		TableInputMeta oldTableInput = new TableInputMeta();
		oldTableInput.setDatabaseMeta(sourcedb);
		oldTableInput.setSQL(oldInputSql);
		StepMeta oldTableInputMetaStep = new StepMeta("table_1_old",oldTableInput);
		transMeta.addStep(oldTableInputMetaStep);

		//表2输入（NEW）
		TableInputMeta newTableInput = new TableInputMeta();
		newTableInput.setDatabaseMeta(sourcedb);
		newTableInput.setSQL(newInputSql);
		StepMeta newTableInputMetaStep = new StepMeta("table_2_new",newTableInput);
		transMeta.addStep(newTableInputMetaStep);

		//合并记录(Merge row)
		MergeRowsMeta mergeRowsMeta = new MergeRowsMeta();
		StepIOMetaInterface stepIOMeta = mergeRowsMeta.getStepIOMeta();
		stepIOMeta.getInfoStreams().get(0).setStepMeta( oldTableInputMetaStep);
		stepIOMeta.getInfoStreams().get(1).setStepMeta( newTableInputMetaStep);
		mergeRowsMeta.setFlagField("diff_type");//设置操作标志字段名
		mergeRowsMeta.setKeyFields(fieldsPrimaryKey);
		mergeRowsMeta.setValueFields(fieldsCompareValue);
		
		StepMeta mergeStepMeta = new StepMeta("merge_diff",mergeRowsMeta);
		transMeta.addStep(mergeStepMeta);
		
		transMeta.addTransHop(new TransHopMeta(oldTableInputMetaStep, mergeStepMeta));
		transMeta.addTransHop(new TransHopMeta(newTableInputMetaStep, mergeStepMeta));

		
		//数据同步
		SynchronizeAfterMergeMeta synchronizeAfterMergeMeta = new SynchronizeAfterMergeMeta();
		synchronizeAfterMergeMeta.setCommitSize(10000); //设置事务提交数量
		synchronizeAfterMergeMeta.setDatabaseMeta(targetdb); //目标数据源
		synchronizeAfterMergeMeta.setSchemaName(targetSchemaName);//数据表schema
		synchronizeAfterMergeMeta.setTableName(targetTableName); //数据表名称
		synchronizeAfterMergeMeta.setUseBatchUpdate(true); //设置批量更新
		synchronizeAfterMergeMeta.setKeyLookup(fieldsPrimaryKey); //设置用来查询的关键字
		synchronizeAfterMergeMeta.setKeyStream(fieldsPrimaryKey); //设置流输入的字段
		
		String[] stream2 = new String [fieldsPrimaryKey.length];
		for(int i = 0; i < fieldsPrimaryKey.length; ++i) {
			stream2[i] = "";
		}
		
		String[] condition = new String [fieldsPrimaryKey.length];
		for (int j = 0; j < fieldsPrimaryKey.length; ++j) {
			condition[j] = "=";
		}
		
		synchronizeAfterMergeMeta.setKeyStream2(stream2);//一定要加上
		synchronizeAfterMergeMeta.setKeyCondition(condition); //设置操作符
		//设置要更新的字段
 		synchronizeAfterMergeMeta.setUpdateLookup(updateLookup);
 		synchronizeAfterMergeMeta.setUpdateStream(updateStream);
 		synchronizeAfterMergeMeta.setUpdate(updateOrNot);
 		
 		//设置高级属性(操作)
 		synchronizeAfterMergeMeta.setOperationOrderField("diff_type"); //设置操作标志字段名
 		synchronizeAfterMergeMeta.setOrderInsert("new");
 		synchronizeAfterMergeMeta.setOrderUpdate("changed");
 		synchronizeAfterMergeMeta.setOrderDelete("deleted");
 		
 		StepMeta synStepMeta = new StepMeta("数据同步", synchronizeAfterMergeMeta);
 		transMeta.addStep(synStepMeta);

		transMeta.addTransHop(new TransHopMeta(mergeStepMeta, synStepMeta));
		
		//打印转换的XML配置
		String transXml = transMeta.getXML();
		System.out.println("=================");
		System.out.println(transXml);
		
		//执行转换过程
		Trans trans = new Trans(transMeta);
		try {
			trans.execute(null);
			System.out.println("start............");
			trans.waitUntilFinished();
			System.out.println("end............");
			if (trans.getErrors() > 0) {
				String errMsg = KettleLogStore.getAppender().getBuffer(trans.getLogChannelId(), false).toString();
				throw new KettleException(errMsg);
			}
		} finally {
			trans.cleanup();
		}
	}

}
